package com.shujia.flink.state

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.{lang, util}

object Demo3ListState {
  def main(args: Array[String]): Unit = {

    /**
     * 实时统计每一个班级的平均年龄
     *
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)
    val kvDS: DataStream[(String, Double)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      val clazz: String = split(4)
      val age: Double = split(2).toDouble
      (clazz, age)
    })

    //安装班级分组
    val keyByDS: KeyedStream[(String, Double), String] = kvDS.keyBy(_._1)

    keyByDS.process(new KeyedProcessFunction[String, (String, Double), (String, Double)] {

      var agesState: ListState[Double] = _

      override def open(parameters: Configuration): Unit = {
        //获取环境
        val context: RuntimeContext = getRuntimeContext

        /**
         * ListState: 集合状态，为每一个key在状态中保存一个集合
         */
        //定义状态
        agesState = context.getListState(new ListStateDescriptor[Double]("ages", classOf[Double]))
      }

      override def processElement(value: (String, Double),
                                  ctx: KeyedProcessFunction[String, (String, Double), (String, Double)]#Context,
                                  out: Collector[(String, Double)]): Unit = {
        val (clazz: String, age: Double) = value

        //没来一个新的学生将年龄保存在集合状态中
        agesState.add(age)

        /**
         * 计算平均年龄
         */
        //获取状态中保存的所有的数据
        val iterable: lang.Iterable[Double] = agesState.get()
        //java集合转换成scala集合-- 徐娅萍导入一个隐式转换
        /* import scala.collection.JavaConversions._
         //将java集合住哪火车scala集合
         val ageList: List[Double] = iterable.toList
         val avgAge: Double = ageList.sum / ageList.length*/

        //总的年龄
        var sumAge = 0.0
        //总人数
        var count = 0
        val iter: util.Iterator[Double] = iterable.iterator()
        while (iter.hasNext) {
          sumAge += iter.next()
          count += 1
        }
        val avgAge: Double = sumAge / count

        //将数据发送到下游
        out.collect((clazz, avgAge))

      }
    }).print()

    env.execute()
  }

}
